package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.flink.streaming.strategy.vehiclepredict.ChargePredictProducer;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.File;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;

/**
 * @author : GW00280745
 * @date : 2023/1/29 18:06
 * @description :
 *
 * // kafka-topics --create --zookeeper 10.31.8.112:2181 --replication-factor 1 --partitions 3 --topic dws_multidimensional_debug_230410
 *     // kafka-topics --delete --zookeeper 10.31.8.112:2181 --topic dws_multidimensional_debug_230410
 *     //      查看group列表  kafka-consumer-groups --bootstrap-server 10.31.8.112:9092 --list
 *     //   查看group消费进度  kafka-consumer-groups --bootstrap-server 10.31.8.112:9092 --describe --group abc
 *     //   kafka-configs --zookeeper 10.31.8.112:2181 --describe --entity-name bms_charge_dwm_detail --entity-type topics
 */
public class ChargePredictDebugDwsTask {
    public static String fromDwdTopic = "bms_charge_dwm_detail";
    private String kafkaServer = "10.31.8.112:9092,10.31.8.113:9092,10.31.8.114:9092";
    public static String resTopic = "dws_multidimensional_debug_230410";

    public static void main(String[] args) throws Exception{
        ChargePredictDebugDwsTask dwsTask = new ChargePredictDebugDwsTask();
//        ParameterTool cmdParams = ParameterTool.fromArgs(args);
//        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//        dwsTask.sourceStd(env);
//        env.execute("dws_multidimensional_debug");
        dwsTask.readFromHive();
    }
    /**
     * @description: source数据标准化，目前只是简单的字段归一，未涉及字段和value的transform
     * @author:  GW00280745
     * @date: 2022/7/4 8:10
     * @param:
     * @param: env
     * @return:
     * @return: org.apache.flink.streaming.api.datastream.DataStream<com.alibaba.fastjson.JSONObject>
     **/
    private DataStream<JSONObject> sourceStd(StreamExecutionEnvironment env) {
        DataStreamSource<String> dataStreamSource = env.addSource(consumer(fromDwdTopic, kafkaServer));
       if (false){
           SingleOutputStreamOperator<String> filter = dataStreamSource.filter((FilterFunction<String>) value -> {
               JSONObject jsonObject = JSONObject.parseObject(value);
               String vin = jsonObject.getString("deviceId");
               String startChargeTime = jsonObject.getString("startChargeTime");
               boolean b    =(StringUtils.equalsIgnoreCase(vin,"LGWEEUA5XNE204086") && StringUtils.equalsIgnoreCase(startChargeTime,"1674987010000")) ||
                       (StringUtils.equalsIgnoreCase(vin,"LGWEEUA56NE202089") && StringUtils.equalsIgnoreCase(startChargeTime,"1674994825000")) ||
                       (StringUtils.equalsIgnoreCase(vin,"LGWEEUA58NE201624") && StringUtils.equalsIgnoreCase(startChargeTime,"1674992661000")) ||
                       (StringUtils.equalsIgnoreCase(vin,"LGWEEUA56NE204019") && StringUtils.equalsIgnoreCase(startChargeTime,"1674988788000")) ||
                       (StringUtils.equalsIgnoreCase(vin,"LGWEEUA55NE201886") && StringUtils.equalsIgnoreCase(startChargeTime,"1674991260000"));
               if (b){
                   return true;
               }
               return false;
//            return true
           });
       }
        DataStream<String> filter = odsEs13Source(dataStreamSource);
        sink2Kafka(filter);
        return null;
    }
    public void sink2Kafka(DataStream dataStream) {
        Properties prodProps = new Properties();
        prodProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        prodProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        prodProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        System.out.println(" 目的地是resTopic  ======= "+resTopic);
        dataStream
                .addSink(new ChargePredictProducer(prodProps, resTopic))
                .name("sink2kafka")
        ;
    }

    private FlinkKafkaConsumer<String> consumer(String dwdTopic, String server){
        String hHmmss = DateFormatUtils.format(System.currentTimeMillis(), "HHmmss");
//        String defaultGroupId = "multidimensional_debug_".concat(hHmmss)
        String defaultGroupId = dwdTopic+"#".concat(hHmmss);
        System.out.println(" server ======= "+server);
        System.out.println(" 消费的源是dwdTopic ======= "+dwdTopic);
        System.out.println(" defaultGroupId ======= [ kafka-consumer-groups --bootstrap-server 10.31.8.112:9092 --describe --group "+defaultGroupId);
        FlinkKafkaConsumer<String> consumer = FlinkKafkaUtil.getConsumer(server, defaultGroupId, dwdTopic);
        consumer.setStartFromEarliest();
//        consumer.setStartFromLatest()
        return consumer;
    }
    private DataStream<String> odsEs13Source(DataStream<String> sourceDataStream) {
        SingleOutputStreamOperator<String> filter = sourceDataStream.filter((FilterFunction<String>) value -> {
            JSONObject line = JSONObject.parseObject(value);
            //充电组名称 如果为空或者-999999 则舍弃
            String bmsDcChrgConnect = line.getString("BMS_DCChrgConnect");
            boolean dcIsOne = StringUtils.equalsIgnoreCase(bmsDcChrgConnect, "1");

            String startChargeTime = line.getString("startChargeTime");
            int compareTo = startChargeTime.compareTo("1672848000000");
            int compareTo2 = startChargeTime.compareTo("1675180800000");
            boolean timeCompare = compareTo >= 0 && compareTo2 <= 0;
            boolean b =  dcIsOne && timeCompare;
            return b;
//            boolean b1 = StringUtils.equalsIgnoreCase(startChargeTime, "1661947458000")
//            return b1
        });
        return filter;
    }

    private void readFromHive(){
        try {
            Properties prodProps = new Properties();
            prodProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
            prodProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            prodProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
            //2M 或 1000ms 进行一次 producer.flush() 负载高用2M负载低用ms
            prodProps.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"23840");
            prodProps.setProperty(ProducerConfig.LINGER_MS_CONFIG,"1000");
            //消息发送失败后重试次数=5
            prodProps.setProperty(ProducerConfig.RETRIES_CONFIG,"5");
            KafkaProducer<String, String> producer = new KafkaProducer<>(prodProps);

            String topic = "dws_multidimensional_predict_debug_230413";
            // starttid, brand, source, bms_chrgsts, bms_battcurr_rm, bms_rmc_moduletempmax, bms_rmc_moduletempmin, bms_dcchrgconnect, obc_connectsts_ev, bms_soc,
            //  soh_cor, cell_battery_type, batter_capcity, bms_heatreq, bms_coolreq, bms_innersoc, bms_innersocmax, bms_innersocmin, bms_cellvoltmax, bms_cellvoltmin,
            //  type, deviceid, item_time, vehicletype, dt, pack_capacity, model_code, bms_chrgtime, bms_soh, startchargetime,
            //  acopensts, acambtemp, aoiid, bms_soe, aoitype, adcode, poiid, poitype, custom_type, chargetype
            List<String> list = FileUtils.readLines(new File("E:\\fuwuqi\\query-hive-13768.csv"), "utf-8");
            String com=",";
            for (String line : list) {
                String[] split = line.split(com, -1);
                JSONObject json = new JSONObject();
                String starttid = "-999999";
                json.put("startTid",starttid);
                String brand = split[0];
                json.put("brand",brand);
                String source = split[1];
                json.put("source",source);
                String bms_chrgsts = split[2];
                json.put("BMS_ChrgSts",bms_chrgsts);
                String bms_battcurr_rm = split[3];
                json.put("BMS_BattCurr_RM",bms_battcurr_rm);
                String bms_rmc_moduletempmax = split[4];
                json.put("BMS_RMC_ModuleTempMax",bms_rmc_moduletempmax);
                String bms_rmc_moduletempmin = split[5];
                json.put("BMS_RMC_ModuleTempMin",bms_rmc_moduletempmin);
                String bms_dcchrgconnect = split[6];
                json.put("BMS_DCChrgConnect",bms_dcchrgconnect);
                String obc_connectsts_ev = split[7];
                json.put("OBC_ConnectSts_EV",obc_connectsts_ev);
                String bms_soc = split[8];
                json.put("BMS_SOC",bms_soc);
                String soh_cor = split[9];
                json.put("SOH_COR",soh_cor);
                String cell_battery_type = split[10];
                json.put("cell_battery_type",cell_battery_type);
                String batter_capcity = split[11];
                json.put("batter_capcity",batter_capcity);
                String bms_heatreq = split[12];
                json.put("BMS_HeatReq",bms_heatreq);
                String bms_coolreq = split[13];
                json.put("BMS_CoolReq",bms_coolreq);
                String bms_innersoc = split[14];
                json.put("BMS_InnerSOC",bms_innersoc);
                String bms_innersocmax = split[15];
                json.put("BMS_InnerSOCmax",bms_innersocmax);
                String bms_innersocmin = split[16];
                json.put("BMS_InnerSOCmin",bms_innersocmin);
                String bms_cellvoltmax = split[17];
                json.put("BMS_CellVoltMax",bms_cellvoltmax);
                String bms_cellvoltmin = split[18];
                json.put("BMS_CellVoltMin",bms_cellvoltmin);
                String type = split[19];
                json.put("type",type);
                String deviceid = split[20];
                json.put("deviceId",deviceid);
                String item_time = split[21];
                json.put("item_time",item_time);
                String vehicletype = split[22];
                json.put("vehicletype",vehicletype);
                String dt = split[23];
                json.put("datedt",dt);
                String pack_capacity = split[24];
                json.put("pack_capacity",pack_capacity);
                String model_code = split[25];
                json.put("model_code",model_code);
                String bms_chrgtime = split[26];
                json.put("BMS_ChrgTime",bms_chrgtime);
                String bms_soh = split[27];
                json.put("BMS_SOH",bms_soh);
                String startchargetime = split[28];
                json.put("startChargeTime",startchargetime);
                String acopensts = split[29];
                json.put("ACOpenSts",acopensts);
                String acambtemp = split[30];
                json.put("ACAmbTemp",acambtemp);
                String aoiid = split[31];
                json.put("aoiId",aoiid);
                String bms_soe = split[32];
                json.put("BMS_SOE",bms_soe);
                String aoitype = split[33];
                json.put("aoiType",aoitype);
                String adcode = split[34];
                json.put("adCode",adcode);
                String poiid = split[35];
                json.put("poiId",poiid);
                String poitype = split[36];
                json.put("poiType",poitype);
                String custom_type = split[37];
                json.put("custom_type",custom_type);
                String chargetype = split[38];
                json.put("chargetype",chargetype);
                ProducerRecord<String, String> record = new ProducerRecord<>(topic,deviceid, json.toString());
                Future<RecordMetadata> future = producer.send(record);
            }
            producer.flush();
            producer.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
